package com.jial08.rabbitmqretry.config;

import com.jial08.rabbitmqretry.domain.QueueConstant;
import com.jial08.rabbitmqretry.domain.QueueEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <p>类描述：Rabbitmq 配置。</p>
 *
 * @author jl
 * @version v1.0.0.1。
 * @since JDK1.8。
 * <p>创建日期：2020/12/8 下午10:25。</p>
 */
@Configuration
@Slf4j
public class RabbitmqConfig {

	@Autowired
	private RabbitTemplate rabbitTemplate;

	@Bean
	public TopicExchange payExchange() {
		return ExchangeBuilder.topicExchange(QueueEnum.QUEUE_ORDER_PAY.getExchange()).durable(true).build();
	}

	@Bean
	public TopicExchange retryExchange() {
		return ExchangeBuilder.topicExchange(QueueEnum.QUEUE_ORDER_PAY_RETRY.getExchange()).durable(true).build();
	}

	@Bean
	public TopicExchange failedExchange() {
		return ExchangeBuilder.topicExchange(QueueEnum.QUEUE_ORDER_PAY_FAILD.getExchange()).durable(true).build();
	}

	@Bean
	public Queue payQueue() {
		return new Queue(QueueEnum.QUEUE_ORDER_PAY.getName(), true);
	}

	@Bean
	public Queue retryQueue() {
		Map args = new ConcurrentHashMap<>(4);
		// 将消息重新投递到 payExchange 中
		args.put(QueueConstant.X_DEAD_LETTER_EXCHANGE, QueueEnum.QUEUE_ORDER_PAY.getExchange());
		args.put(QueueConstant.X_DEAD_LETTER_ROUTING_KEY, QueueEnum.QUEUE_ORDER_PAY_DEAD.getRouteKey());
		//消息在队列中延迟30s后超时，消息会重新投递到 x-dead-letter-exchage 对应的队列中，routingKey 为自己指定
		args.put(QueueConstant.X_MESSAGE_TTL, QueueConstant.RETRY_SECOND);
		return QueueBuilder.durable(QueueEnum.QUEUE_ORDER_PAY_RETRY.getName()).withArguments(args).build();
	}

	@Bean
	public Queue failedQueue() {
		return QueueBuilder.durable(QueueEnum.QUEUE_ORDER_PAY_FAILD.getName()).build();
	}

	@Bean
	public Binding topicQueueBinding(@Qualifier("payQueue") Queue queue,
	                                 @Qualifier("payExchange") TopicExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with("mall.pay.topic.*");
	}
	@Bean
	public Binding retryQueueBinding(@Qualifier("retryQueue") Queue queue,
	                                  @Qualifier("retryExchange") TopicExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with("mall.pay.retry.*");
	}
	@Bean
	public Binding failQueueBinding(@Qualifier("failedQueue") Queue queue,
	                                 @Qualifier("failedExchange") TopicExchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with("mall.pay.failed.*");
	}

	@Bean
	public AmqpTemplate amqpTemplate(){
		// 消息确认，需要配置 spring.rabbitmq.publisher-confirm-type = correlated
		rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
			// 根据返回的状态，生产者可以处理失败与成功的相应信息，比如发送失败，可重发，转发或者存入日志等
			// correlationData.getId()为 message 唯一标识，需要生产者发送message时传入自定义的correlationData才能获取到，否则为 null
			if(!ack){
				String id = correlationData.getId();
				log.warn("消息消费失败：{}，原因：{}", id, cause);
			}
		});

		// 消息发送失败返回到队列中，需要配置spring.rabbitmq.publisher-returns = true
//		rabbitTemplate.setReturnsCallback(returnedMessage -> log.warn("消息发送失败：{}", returnedMessage));
		// 高版本废弃方法，如果上面方法报错可以用下面的
		rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) ->
				log.warn("消息发送失败：{}", message));
		return rabbitTemplate;
	}

}
